/**
 * 
 */
package org.perfect.batch.step;

import org.perfect.batch.partition.RemotePartitionHandler;
import org.perfect.batch.partition.RemoteStepExecutionAggregator;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.partition.support.Partitioner;
import org.springframework.batch.core.partition.support.SimplePartitioner;
import org.springframework.beans.factory.annotation.Autowired;

/**
 * @author billmse
 *
 */
public class PartitionMasterStep {

	@Autowired
	private StepBuilderFactory stepBuilderFactory;

	@Autowired
	private RemotePartitionHandler partitionHandler;

	@Autowired
	private RemoteStepExecutionAggregator aggregator;

	public Step build(String masterStepName, String slaveStepName) {
		return build(masterStepName, slaveStepName, null);
	}

	public Step build(String masterStepName, String slaveStepName, Partitioner partitioner) {
		partitioner = (null == partitioner ? new SimplePartitioner() : partitioner);
		return stepBuilderFactory.get(masterStepName).partitioner(slaveStepName, partitioner)
				.partitionHandler(partitionHandler).aggregator(aggregator).build();
	}

}
